package stateful;

import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.queryablestate.client.QueryableStateClient;

import java.io.IOException;
import java.util.concurrent.CompletableFuture;

/**
 * 状态查询客户端
 */
public class QueryableStateClientDemo {

    public static void main(String[] args) {
        // 创建查询状态客户端 指定主机 端口
        QueryableStateClient client = null;
        try {
            client = new QueryableStateClient("localhost", 9069);
        } catch (Exception e){
            e.printStackTrace();
        }
        // 创建状态描述信息
        ValueStateDescriptor<Integer> stateDescriptor = new ValueStateDescriptor<>("count-state", Integer.class);
        // 创建job id 对象
        JobID jobID = JobID.fromHexString("62c32de6da54c0581e77d8737f2a66a0");
        // 获取状态 指定job id 状态名称 key key的类型 状态描述信息
        CompletableFuture<ValueState<Integer>> kvState = client.getKvState(jobID, "wc", "spark", BasicTypeInfo.STRING_TYPE_INFO, stateDescriptor);

        kvState.thenAccept(resp -> {
            try {
                Integer value = resp.value();
                System.out.println(value);
            } catch (IOException e) {
                e.printStackTrace();
            }
        });

        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }
}
